import akka.actor.*;
import akka.routing.RoundRobinPool;
import java.util.concurrent.CountDownLatch;
import java.util.Random;

/**
 * 表示工作任务的消息类。
 * 每个任务由一个唯一的 taskId 标识。
 *
 * @param taskId 任务的唯一标识符
 */
record WorkMessage(int taskId) {
}

/**
 * 表示任务执行结果的消息类。
 * 包含执行该任务的 actorId、任务 ID 和执行结果字符串。
 *
 * @param actorId 执行任务的 Actor 的 ID
 * @param taskId 任务的唯一标识符
 * @param result 任务执行的结果描述
 */
record ResultMessage(int actorId, int taskId, String result) {
}

/**
 * 表示启动消息的类，用于通知 ActiveObjectActor 开始工作。
 *
 * @param latch 用于同步所有 Actor 启动完成的 CountDownLatch
 */
record StartMessage(CountDownLatch latch) {
}

/**
 * 表示随机消息的类，用于 Actor 之间互相通信。
 *
 * @param senderId 发送方 Actor 的 ID
 * @param content 消息内容
 */
record RandomMessage(int senderId, String content) {
}

/**
 * 表示活动对象的 Actor 类。
 * 该类负责处理工作任务、随机消息，并维护自身状态。
 */
class ActiveObjectActor extends AbstractActor {
    private final int actorId;
    private int completedTasks = 0;
    private ActorRef router; // 保存路由器引用，用于发送随机消息
    private final Random random = new Random();

    /**
     * 计数器变量，用于记录某个操作的执行次数
     * 该变量被声明为volatile以确保多线程环境下的可见性
     * 静态变量保证在类的所有实例间共享同一个计数器
     */
    private static volatile long counter = 0;

    /**
     * 构造函数，初始化 ActiveObjectActor 实例。
     *
     * @param actorId 当前 Actor 的唯一标识符
     */
    public ActiveObjectActor(int actorId) {
        this.actorId = actorId;
        counter++ ;
    }

    /**
     * 定义当前 Actor 的消息处理行为。
     * 支持处理 WorkMessage、StartMessage 和 RandomMessage 类型的消息。
     *
     * @return Receive 消息处理器
     */
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(WorkMessage.class, this::handleWork)
                .match(StartMessage.class, this::handleStart)
                .match(RandomMessage.class, this::handleRandomMessage)
                .build();
    }

    /**
     * 处理 StartMessage 消息。
     * 在收到启动消息后，执行内部工作并通知计数门闩。
     *
     * @param msg 启动消息对象
     */
    private void handleStart(StartMessage msg) {
        // 保存路由器引用
        this.router = getContext().getParent();

        // 收到启动消息后开始工作
        for (int i = 0; i < 10; i++) {
            // 模拟内部工作
            doWork(i);
        }
        msg.latch().countDown();
    }

    /**
     * 处理 WorkMessage 消息。
     * 执行指定的任务，并向发送者返回结果。
     * 同时有一定概率向其他 Actor 发送随机消息。
     *
     * @param msg 工作任务消息对象
     */
    private void handleWork(WorkMessage msg) {
        // 处理工作消息
        String result = doWork(msg.taskId());
        // 发送结果给发送者
        getSender().tell(new ResultMessage(actorId, msg.taskId(), result), getSelf());

        // 随机向其他Actor发送消息
        if (router != null && random.nextInt(100) < 10) { // 10%概率发送随机消息

            router.tell(new RandomMessage(actorId, "Hello from Actor " + actorId), getSelf());
        }
    }

    /**
     * 处理来自其他 Actor 的 RandomMessage 消息。
     * 有一定概率回复发送方，并模拟处理时间。
     *
     * @param msg 随机消息对象
     */
    private void handleRandomMessage(RandomMessage msg) {
        // 处理来自其他Actor的随机消息
        // 可以选择回复或者忽略
        if (random.nextInt(100) < 5) { // 5%概率回复
            getSender().tell(new RandomMessage(actorId, "Reply from Actor " + actorId + " to " + msg.content()), getSelf());
        }
        if(actorId == 0){
            System.out.println("Actor 0 received a random message from Actor " + msg.senderId());
        }
        // 模拟处理时间
        try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 执行具体的工作任务。
     * 增加已完成任务计数，并模拟计算时间。
     *
     * @param taskId 当前任务的 ID
     * @return 描述任务完成情况的字符串
     */
    private String doWork(int taskId) {
        completedTasks++;
        // 模拟实际工作
        try {
            Thread.sleep(1); // 模拟计算时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return String.format("Actor %d completed task %d (total: %d)",
                actorId, taskId, completedTasks);
    }
}

/**
 * 负责收集任务执行结果的 Actor。
 * 统计接收到的结果数量和随机消息数量，并在所有任务完成后终止系统。
 */
class ResultCollector extends AbstractActor {
    private int totalResults = 0;
    private final long startTime;
    private int randomMessages = 0; // 记录收到的随机消息数量

    /**
     * 构造函数，初始化 ResultCollector 实例。
     *
     * @param startTime 程序开始执行的时间戳（毫秒）
     */
    public ResultCollector(long startTime) {
        this.startTime = startTime;
    }

    /**
     * 定义当前 Actor 的消息处理行为。
     * 支持处理 ResultMessage 和 RandomMessage 类型的消息。
     *
     * @return Receive 消息处理器
     */
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(ResultMessage.class, this::handleResult)
                .match(RandomMessage.class, this::handleRandomMessage)
                .build();
    }

    /**
     * 处理任务执行结果消息。
     * 统计已接收的结果数量，并在所有任务完成后输出统计信息并终止系统。
     *
     * @param msg 任务执行结果消息对象
     */
    private void handleResult(ResultMessage msg) {
        totalResults++;
        // 每10000个结果打印一次进度
        if (totalResults % 10000 == 0) {
            System.out.printf("Received %d results so far...%n", totalResults);
        }

        // 所有任务完成时
        if (totalResults == 1000000) { // 100,000 actors * 10 tasks each
            long endTime = System.currentTimeMillis();
            System.out.printf("All %d tasks completed in %d ms%n",
                    totalResults, endTime - startTime);
            System.out.printf("Received %d random messages%n", randomMessages);
            // 关闭系统
            getContext().getSystem().terminate();
        }
    }

    /**
     * 处理来自其他 Actor 的随机消息。
     * 统计已接收的随机消息数量，并定期输出统计信息。
     *
     * @param msg 随机消息对象
     */
    private void handleRandomMessage(RandomMessage msg) {
        randomMessages++;
        // 每1000个随机消息打印一次
        if (randomMessages % 1000 == 0) {
            System.out.printf("Received %d random messages, latest: %s%n",
                    randomMessages, msg.content());
        }
    }
}

/**
 * 主类，用于启动整个 Akka 活动对象系统。
 * 创建大量 ActiveObjectActor 并分发工作任务，最后收集结果。
 */
public class AkkaActiveObjects {
    /**
     * 程序入口点。
     * 初始化 Actor 系统，创建路由器和结果收集器，
     * 分发启动和工作任务消息，并等待所有任务完成。
     *
     * @param args 命令行参数（未使用）
     * @throws InterruptedException 如果线程被中断
     */
    public static void main(String[] args) throws InterruptedException {
        final int ACTOR_COUNT = 1_000_000; // 十万个活动对象
        long startTime = System.currentTimeMillis();

        // 创建Actor系统
        ActorSystem system = ActorSystem.create("ActiveObjectsSystem");

        // 创建结果收集器
        ActorRef resultCollector = system.actorOf(
                Props.create(ResultCollector.class, startTime),
                "resultCollector"
        );

        // 创建一个路由器来管理我们的活动对象
        // 使用RoundRobinPool进行负载均衡
        ActorRef workerRouter = system.actorOf(
                new RoundRobinPool(ACTOR_COUNT)
                        .props(Props.create(ActiveObjectActor.class, 0)), // 临时ID，后面会重写
                        "workerRouter"
        );

        // 等待所有Actor启动的计数器
        CountDownLatch startLatch = new CountDownLatch(ACTOR_COUNT);

        System.out.println("Starting " + ACTOR_COUNT + " active objects...");

        // 启动所有活动对象
        for (int i = 0; i < ACTOR_COUNT; i++) {
            // 修改ActiveObjectActor的构造方式，传递不同的actorId
            ActorRef actor = system.actorOf(Props.create(ActiveObjectActor.class, i), "actor" + i);
            
            // 为每个Actor发送启动消息
            actor.tell(new StartMessage(startLatch), resultCollector);

            // 发送10个工作任务给每个Actor
            for (int j = 0; j < 10; j++) {
                actor.tell(new WorkMessage(j), resultCollector);
            }
        }

        // 等待所有Actor启动完成
        startLatch.await();
        System.out.println("All active objects started in " +
                (System.currentTimeMillis() - startTime) + " ms");
    }
}
